package com.db.concurrency.producerconsumer.executors;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;

/**
 * @author Alexander Kotov (kotov.alex.22@gmail.com)
 */
public class FileIndexer implements Runnable {

    private final int id;
    private final List<BlockingDeque<File>> queueList;
    private final BlockingDeque<File> queue;
    private final ExecutorService executorService;
    private int indexed;
    private int stolen;
    private int indexedAfterInterruption;

    public FileIndexer(int id, List<BlockingDeque<File>> queueList, ExecutorService executorService) {
        this.id = id;
        this.queue = queueList.get(id);
        this.queueList = queueList;
        this.executorService = executorService;
    }

    public void run() {
        File file = queue.pollFirst();
        if (file != null) {
            indexFile(file);
            indexed++;
        } else {
            int curId = id;
            for (int i = 0; i < queueList.size() - 1; ++i) {
                curId = ++curId % queueList.size();
                file = queueList.get(curId).pollLast();
                if (file != null) {
                    indexFile(file);
                    stolen++;
                    break;
                }
            }
        }

        if (Thread.currentThread().isInterrupted()) {
            while ((file = queue.pollLast()) != null) {
                indexFile(file);
                indexed++;
                indexedAfterInterruption++;
            }
            System.out.format("FileIndexer %d: indexed %d (after interruption %d) files, stolen %d files%n",
                    id, indexed, indexedAfterInterruption, stolen);
            Thread.currentThread().interrupt();
        } else {
            executorService.submit(this);
        }
    }

    public void indexFile(File file) {
        int linesCount = 0;
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                linesCount++;
            }
            System.out.format("%s: File '%s' has been indexed, it contains %d lines%n",
                    Thread.currentThread().getName(), file.getAbsolutePath(), linesCount);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
    }
}

